package com.zyx.flinkdemo.stream.statebackend;


import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author zyx
 * @since 2021/8/14 00:09
 * desc: RocksDBStateBackend案例
 */
public class RocksdbStateBackendDemo {
    public static void main(String[] args) throws Exception {
        // 获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 允许检查点
        env.enableCheckpointing(10 * 1000L);
        // 设置RocksDBStateBackend为状态后端, true表示使用增量快照
        EmbeddedRocksDBStateBackend rocksdbBackend = new EmbeddedRocksDBStateBackend(true);
        // 自定义RocksDBBackend的配置
        /*rocksdbBackend.setRocksDBOptions(new MyOptionsFactory());*/
        env.setStateBackend(rocksdbBackend);
        env.getCheckpointConfig().setCheckpointStorage("hdfs://linux203:8020/test/checkpoint_test");
        // 读取数据源数据并打印
        env.socketTextStream("linux201", 7777).print();
        // 执行环境
        env.execute();
    }
}
